package demo.kafka;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

import java.util.Properties;

public class KafkaProducerDemo {
    public static final String KFKADD = "10.10.15.243:9092,10.10.15.224:9092,10.10.15.222:9092";

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /** 无法测试，本地没法访问集群地址 仅供api使用参考 */
        DataStreamSource<String> input = env.socketTextStream("localhost", 9999);
        String sinkTopic = "test";
        Properties producerProp = new Properties();
        producerProp.setProperty("bootstrap.servers", KFKADD);

        FlinkKafkaProducer<String> stringFlinkKafkaProducer =
                new FlinkKafkaProducer<String>(sinkTopic, new SimpleStringSchema(), producerProp);
        input.addSink(stringFlinkKafkaProducer);

        env.execute("flink-producer-demo");
    }
}
